package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Component
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        stream.map((key, value) -> {
            System.out.println("key : " + key + " value: " +value);
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd()); // id:like:1
        })
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始化参数的方法-每个时间创建开始的时候执行一次
                     * 作用：完成统计数据的初始化和上一次统计结果的归0
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * 统计每个消息中的数据
                     * 每次消息来都会执行一次
                     * @param key KeyStream中的key - 文章的id
                     * @param value KeyStream中的value - like:1
                     * @param aggregate
                     *  如果是第一个消息那么aggregate就是初始化的值  COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                     *  如果不是第一个消息那么aggregate就是上一次统计的结果 COLLECTION:1,COMMENT:0,LIKES:0,VIEWS:0
                     * @return
                     */
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        //1. 判断value是否为空 如果为空则返回上一次的结果
                        if(StringUtils.isBlank(value)){
                            return aggregate;
                        }
                        //2. 把上一次的结果数据转为4个变量分别存储上一次统计的结果数据
                        int col = 0,com=0,lik=0,vie=0;
                        String[] strs = aggregate.split(","); // [COLLECTION:0,COMMENT:0]
                        for (String str : strs) { // COLLECTION:0
                            String[] values = str.split(":"); //[COLLECTION, 0]
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])){
                                case COLLECTION:
                                    col = Integer.parseInt(values[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(values[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(values[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(values[1]);
                                    break;
                            }
                        }

                        //3. 根据value来实现结果的累加操作 - like:1
                        String[] msgValues = value.split(":");//[like,1]

                        switch (UpdateArticleMess.UpdateArticleType.valueOf(msgValues[0])){
                            case COLLECTION:
                                col += Integer.parseInt(msgValues[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(msgValues[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(msgValues[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(msgValues[1]);
                                break;
                        }

                        //4. 数据格式的转换-COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id:"+key);
                        System.out.println("当前时间窗口内的消息处理结果："+formatStr);

                        return formatStr;
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key, value) -> {
                    System.out.println("key : " + key + " value: " +value);
                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(),value));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }

    /**
     * 把结果转换为一个json数据再发送到kafka
     * @param id
     * @param value
     * @return
     */
    private String formatObj(String id, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();

        mess.setArticleId(Long.parseLong(id));

        String[] strs = value.split(","); // [COLLECTION:0,COMMENT:0]
        for (String str : strs) { // COLLECTION:0
            String[] values = str.split(":"); //[COLLECTION, 0]
            switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(values[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(values[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(values[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(values[1]));
                    break;
            }
        }

        return JSON.toJSONString(mess);
    }
}
